package com.shujia.flink.core

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor

object Demo6WaterMark2 {
  def main(args: Array[String]): Unit = {


    /**
      * 1,1574666000
      * 1,1574671000
      * 1,1574667000
      * 1,1574668000
      * 1,1574669000
      * 1,1574670000
      * 1,1574671000
      * 1,1574672000
      * 1,1574673000
      * 1,1574674000
      * 1,1574675000
      * 1,1574680000
      * 1,1574760000
      *
      */

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //当所有线程中时间到达窗口的结束时间才开始计算
    env.setParallelism(1)
    //处理时间 TimeCharacteristic.ProcessingTime
    //TimeCharacteristic.EventTime   时间时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val ds = env.socketTextStream("node1", 8888)

    val eventDS = ds.map(line => {
      val split = line.split(",")

      Event2(split(0), split(1).toLong)
    })

    //统计最近5秒每个车出现的次数

    eventDS

      //字段时间列 和最大允许的延迟时间
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[Event2](Time.seconds(5)) {
        override def extractTimestamp(element: Event2): Long = {
          element.time
        }
      })


      .map(event => (event.id, 1))
      .keyBy(_._1)
      .timeWindow(Time.seconds(5))
      .reduce((x, y) => (x._1, x._2 + y._2))
      .print()

    env.execute("Demo4EventTime")


  }

  case class Event2(id: String, time: Long)

}
